package com.example.zhouwei.rx2_android;

import android.nfc.Tag;
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;


import java.io.Serializable;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableOperator;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.internal.operators.observable.ObservableGroupBy;
import io.reactivex.observables.GroupedObservable;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.BehaviorSubject;

public class MainActivity extends AppCompatActivity {

    private static final String TAG = "RXAndroid2";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
//        testBuffer();

        Flowable.range(1, 12).onBackpressureDrop().subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.v(TAG, String.valueOf(integer));
            }
        });
    }

    private void testBuffer() {
        Observable.range(0, 10).buffer(new Observable<List<Integer>>() {
            /**
             * Operator implementations (both source and intermediate) should implement this method that
             * performs the necessary business logic.
             * <p>There is no need to call any of the plugin hooks on the current Observable instance or
             * the Subscriber.
             *
             * @param observer the incoming Observer, never null
             */
            @Override
            protected void subscribeActual(Observer<? super List<Integer>> observer) {
                Log.v(TAG, "subscribeActual");
                List<Integer> list =  new ArrayList<>();
                list.add(1111);
                list.add(1232);
                observer.onNext(list);
                observer.onComplete();
            }
        }).subscribe(this.<List<Integer>>getObserver());
    }

    private void testGroupBy() {
//        Observable.range(0, 10).subscribe(this.<Integer>getObserver());
        Observable.just("a", "b", "c", "d", "a").groupBy(new Function<String, String>() {
            @Override
            public String apply(String s) throws Exception {
                return s;
            }
        }).subscribe(new Observer<GroupedObservable<String, String>>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(GroupedObservable<String, String> stringStringGroupedObservable) {
                Log.v(TAG, "key=>"+stringStringGroupedObservable.getKey());
                stringStringGroupedObservable.subscribe(MainActivity.this.<String>getObserver());
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
//        Observable.range(0, 10).groupBy(new Function<Integer, Integer>() {
//            @Override
//            public Integer apply(Integer integer) throws Exception {
//                return integer % 5;
//            }
//        }).subscribe(new Observer<GroupedObservable<Integer, Integer>>() {
//            @Override
//            public void onSubscribe(Disposable d) {
//
//            }
//
//            @Override
//            public void onNext(GroupedObservable<Integer, Integer> integerIntegerGroupedObservable) {
//                integerIntegerGroupedObservable.subscribe(MainActivity.this.<Integer>getObserver());
//            }
//
//            @Override
//            public void onError(Throwable e) {
//
//            }
//
//            @Override
//            public void onComplete() {
//
//            }
//        });
    }

    private void testMap() {
//        Observable.just("abc", "def", "gh").map(new Function<String, String>() {
//            @Override
//            public String apply(String s) throws Exception {
//                return s.substring(0, 1);
//            }
//        }).subscribe(this.<String>getObserver());
        Observable.just("abc", "def", "gh").flatMap(new Function<String, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(final String s) throws Exception {
                return new ObservableSource<String>() {
                    @Override
                    public void subscribe(Observer<? super String> observer) {
                        for (int i=0; i<s.length(); i++) {
                            observer.onNext(s.substring(i, i+1));
                        }
                    }
                };
            }
        }).subscribe(this.getObserver());
    }

    private void testCombine() {
        Observable<Integer> ob1 = Observable.just(1,2,3);
        Observable<Integer> ob2 = Observable.just(6);

//        Observable.combineLatest(ob1, ob2, new BiFunction<Integer, Integer, Integer>() {
//            @Override
//            public Integer apply(Integer integer, Integer integer2) throws Exception {
//                Log.v(TAG, "integer=>" + String.valueOf(integer) + "integer2=>" + String.valueOf(integer2));
//                return integer + integer2;
//            }
//        }).subscribe(this.<Integer>getObserver());
        List<Integer> list = new ArrayList<>();
        list.add(11);
        list.add(12);
        ob1.startWith(list).subscribe(this.<Integer>getObserver());
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        bs.onComplete();
    }


    BehaviorSubject<String> bs = BehaviorSubject.create();

    private void testBehaviorSubject() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                for (int i=0; ; i++) {
                    Thread.sleep(1000);
                    e.onNext(String.valueOf(i));
                }
            }
        }).takeUntil(bs).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(this.<String>getObserver());
    }

    private void testTakeUtil() {
//        Observable.just("11111", "22222222").takeLast(1).subscribe(this.<String>getObserver());
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                for (int i=0; ; i++) {
                    Thread.sleep(1000);
                    e.onNext(String.valueOf(i));
                }
            }
        }).takeUntil(Observable.interval(5, TimeUnit.SECONDS)).observeOn(Schedulers.io()).subscribeOn(AndroidSchedulers.mainThread()).subscribe(this.<String>getObserver());;
    }

    private void testCompose() {
        Observable.just("11111", "222222", "33").compose(new ObservableTransformer<String, Long>() {
            @Override
            public ObservableSource<Long> apply(final Observable<String> upstream) {
                return upstream.map(new Function<String, Long>() {
                    @Override
                    public Long apply(String s) throws Exception {
                        return Long.valueOf(s);
                    }
                });
            }
        }).subscribe(this.<Long>getObserver());
    }

    private void testIntervalRange() {
        Observable.intervalRange(0L, 10L, 0L, 3000L, TimeUnit.MILLISECONDS).subscribe(this.<Long>getObserver());
    }

    private void testInterval() {
//        Observable.interval(5, TimeUnit.SECONDS).map(new Function<Long, String>() {
//            @Override
//            public String apply(Long aLong) throws Exception {
//                Log.v(TAG, "map:" + aLong);
//                return String.valueOf(aLong);
//            }
//        }).subscribe(this.<String>getObserver());
        Observable.interval(3, TimeUnit.SECONDS).lift(new ObservableOperator<String, Long>() {
            @Override
            public Observer<? super Long> apply(final Observer<? super String> observer) throws Exception {
                return new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        observer.onSubscribe(d);
                    }

                    @Override
                    public void onNext(Long aLong) {
                        Log.v(TAG, "lift=>" + aLong);
                        observer.onNext(String.valueOf(aLong));
                    }

                    @Override
                    public void onError(Throwable e) {
                        observer.onError(e);
                    }

                    @Override
                    public void onComplete() {
                        observer.onComplete();
                    }
                };
            }
        }).subscribe(this.<String>getObserver());
    }

    private void testThrow() {
        Observable.error(new Callable<Throwable>() {
            @Override
            public Throwable call() throws Exception {
                return new Throwable("testThrow error");
            }
        }).subscribe(this.getObserver());
    }

    private void testNever() {
        Observable.never().subscribe(this.getObserver());
    }

    private void testEmpty() {
        Observable.empty().subscribe(this.getObserver());
    }

    private void testDefer() {
        Observable.defer(new Callable<ObservableSource<?>>() {
            @Override
            public ObservableSource<?> call() throws Exception {
                return Observable.just("testDefer");
            }
        }).subscribe(this.getObserver());
    }

    private void testFrom() {

//        Observable.fromPublisher(new Publisher<String>() {
//            @Override
//            public void subscribe(Subscriber<? super String> s) {
//                s.onNext("fromPublisher 11111111");
//            }
//        }).subscribe(this.<String>getObserver());
//        Observable.fromArray("ssss").subscribe(this.<String>getObserver());
//        Observable.fromCallable(new Callable<String>() {
//            @Override
//            public String call() throws Exception {
//                return "135435335";
//            }
//        }).subscribe(this.<String>getObserver());
//        Observable.fromFuture(new Future<String>() {
//            @Override
//            public boolean cancel(boolean mayInterruptIfRunning) {
//                return false;
//            }
//
//            @Override
//            public boolean isCancelled() {
//                return false;
//            }
//
//            @Override
//            public boolean isDone() {
//                return false;
//            }
//
//            @Override
//            public String get() throws InterruptedException, ExecutionException {
//                return "fsfdsfsdfdsfsfsdf";
//            }
//
//            @Override
//            public String get(long timeout, @NonNull TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
//                return "1111111111111111111";
//            }
//        }).subscribe(this.<String>getObserver());
    }

    private void testJust() {
        Observable.just("1111", "2222", "aaaa").subscribe(this.<String>getObserver());
    }

    private void testCreate() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("111");
                e.onNext("222");
                e.onNext("333");
                e.onComplete();
            }
        }).subscribe(this.<String>getObserver());
    }

    private <T> Observer<T> getObserver (){
        return new Observer<T>() {
            @Override
            public void onSubscribe(Disposable d) {
                if (d.isDisposed()) {
                    Log.v(TAG, "isDisposed" + " is true");
                    d.dispose();
                } else {
                    Log.v(TAG, "isDisposed" + " is false");
                }
            }

            @Override
            public void onNext(T s) {
                Log.v(TAG, "onNext" + s.toString());
            }

            @Override
            public void onError(Throwable e) {
                Log.v(TAG, "onError" + e.getMessage());
            }

            @Override
            public void onComplete() {
                Log.v(TAG, "onComplete");
            }
        };
    }
}
